package com.rui.rabbitmq.four;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rui.utils.RabbitMqUtil;

import java.util.UUID;

public class ConfirmMessage {

    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        //单个确认
//        publishMessageIndividual();
        //批量确认
//        publishMessageBathch();
        //异步确认
        publishMessageAsync();
    }

    //单个确认
    public static void publishMessageIndividual() throws Exception {
        Connection connection = RabbitMqUtil.getNewConnection();
        Channel channel = connection.createChannel();

        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes("UTF-8"));
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println("消息发送成功");
            }
        }

        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "条单独确认消息，耗时：" + (end-begin) + "毫秒");
    }

    //批量发布确认

    public static void publishMessageBathch() throws Exception{
        Connection connection = RabbitMqUtil.getNewConnection();
        Channel channel = connection.createChannel();

        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        //批量确认消息的大小
        int batchSize = 100;

        //批量发送消息 批量发布确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
            if((i+1) % batchSize == 0){
                channel.waitForConfirms();
            }
        }

        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "条批量确认消息，耗时：" + (end-begin) + "毫秒");
    }

    public static void publishMessageAsync() throws Exception {
        Connection connection = RabbitMqUtil.getNewConnection();
        Channel channel = connection.createChannel();

        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);

        //开启发布确认
        channel.confirmSelect();

        //开始时间
        long begin = System.currentTimeMillis();

        //成功的回调
        ConfirmCallback ackCallback = (deliverTag , multiple) -> {
            System.out.println("确认的消息：" + deliverTag);
        };

        //失败的回调
        ConfirmCallback nackCallback = (deliverTag , multiple) -> {
            System.out.println("未确认的消息：" + deliverTag);
        };

        //准备消息的监听器，那些消息成功了，那些消息失败了
        channel.addConfirmListener(ackCallback,nackCallback);

        //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("发布" + MESSAGE_COUNT + "条批量确认消息，耗时：" + (end-begin) + "毫秒");
    }

}
